package com.shujia.kafka

import java.util
import java.util.Properties
import org.apache.kafka.clients.consumer.{ConsumerRecords, KafkaConsumer}


object Demo2Consumer {
  def main(args: Array[String]): Unit = {

    /**
      * 消费者
      *
      */

    val properties: Properties = new Properties

    //group 代表一个消费组   一条数据在一个组内只会消费一次
    properties.put("group.id", "asdadasdaasdsasdafasd")

    //指定kafka broker列表
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")

    //key value  反序列化类
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")

    //从哪里读数据
    /**
      * earliest
      * 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
      * latest
      * 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
      *
      */
    properties.put("auto.offset.reset", "earliest")



    //zk连接超时时间
    properties.put("zookeeper.session.timeout.ms", "10000")

    //zk同步时间
    properties.put("zookeeper.sync.time.ms", "200")

    //自动提交偏移量间隔时间
    properties.put("auto.commit.interval.ms", "1000")


    val consumer: KafkaConsumer[String, String] = new KafkaConsumer[String, String](properties)


    //1、订阅消息
    consumer.subscribe(util.Arrays.asList("student1"))


    while (true) {

      println("等待接收数据")
      //获取数据，指定超时间
      val records: ConsumerRecords[String, String] = consumer.poll(1000)


      //将java集合转换成scala集合需要导入的隐式转换
      import scala.collection.JavaConversions._

      records
        .iterator()
        .toList
        .foreach(record => {

          val key: String = record.key()
          val value: String = record.value()
          val partition: Int = record.partition()
          val topic: String = record.topic()
          val ts: Long = record.timestamp()

          println(s"$key\t$value\t$partition\t$topic\t$ts")

        })

    }

  }
}
